package com.raymond.cmpp.client;

import com.raymond.cmpp.client.msg.MessageManages;
import com.raymond.cmpp.client.netty.ExponentialBackOffRetry;
import com.raymond.cmpp.client.netty.RetryPolicy;
import com.raymond.cmpp.client.queue.PackBlockingQueue;
import com.raymond.cmpp.client.service.Callback;
import com.raymond.cmpp.client.thread.PackRequestThread;
import com.raymond.cmpp.common.msg.BaseMessage;

/**
 * 支持打包的客户端
 *
 * @author :  raymond
 * @version :  V1.0
 * @date :  2022-02-24 15:32
 */
public class PackCmppClient extends CmppClient {

    private int packCount = 10000;

    private final PackBlockingQueue packTask = new PackBlockingQueue();

    public PackCmppClient(String host, int port, String sourceAddr, String password) {
        this(host, port, sourceAddr, password, 0X30);
    }

    public PackCmppClient(String host, int port, String sourceAddr, String password, int version) {
        this(host, port, sourceAddr, password, version, 10000);
    }

    public PackCmppClient(String host, int port, String sourceAddr, String password, int version, int timeout) {
        this(host, port, sourceAddr, password, version, timeout, new ExponentialBackOffRetry(1000, Integer.MAX_VALUE, 60 * 1000));
        packThreadInit();
    }

    public PackCmppClient(String host, int port, String sourceAddr, String password, int version, int timeout, RetryPolicy retryPolicy) {
        super(host, port, sourceAddr, password, version, timeout, retryPolicy);
        packThreadInit();
    }

    private void packThreadInit() {
        PackRequestThread packRequestThread = new PackRequestThread(this.packTask, this, packCount);
        threads.add(packRequestThread);
        packRequestThread.start();
    }

    public void asyncPackSend(BaseMessage req, Callback callback) {
        MessageManages messageManages = new MessageManages(req, callback, this.getTimeout());
        msgManager.put(req.getSequenceId(), messageManages);
        packTask.add(req);
        if (packTask.size() > packCount) {
            packTask.notify0();
            try {
                packTask.wait0(10);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        asyncTask.add(messageManages);
    }
}
